/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package Examples;

import java.util.ArrayList;
import org.apache.hadoop.io.LongWritable;
import utility.PairT;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.Date;
import java.util.Random;
import java.util.Scanner;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 *
 * @author takeshi
 */
public class MpsSeq {

    private static String inputfilename = null, outputfilename = null;
    private static String fEncoding = "UTF-8";
    private static int iRandomsize = -1;
    private static int iBuffsize = 65536;

    public static PairT<Long> mps_sum(ArrayList<Long> xs) {
        Long mps = 0l;
        Long sum = 0l;
        for (Long v : xs) {
            sum += v;
            if (sum > mps) {
                mps = sum;
            }
        }
        return new PairT<Long>(mps, sum);
    }

    public static PairT<Long> OPlus(PairT<Long> f, PairT<Long> b) {

        if (f == null && b == null) {
            return null;
        }
        if (f == null && b != null) {
            return b;
        }
        if (b == null && f != null) {
            return f;
        }
        //  LongPair rst = new LongPair(0, 0);
        long first = 0, second = 0;
        first = Math.max(Math.max(f.getFirst(), f.getSecond() + b.getFirst()), 0);
        second = f.getSecond() + b.getSecond();
        PairT<Long> rst = new PairT<Long>(first, second);
        return rst;
    }

    public static long ComputeMPS(Configuration conf, FileSystem outfs, Path inputpath) throws IOException {

       // long readnum = 0l;
        //     InDATA = new ArrayList<String>();
        // assume the input file is not too large...
        // because it is just for test the correctness
        //byte[] buffer = new byte[4096 * 64];
        //  FileSystem fs = FileSystem.get(URI.create(datafile), conf);
        //  Path inpath = new Path(datafile);
        //  FSDataInputStream fin = fs.open(inpath,4096*64);

        SequenceFile.Reader reader = new SequenceFile.Reader(outfs, inputpath, conf);

        LongWritable key = new LongWritable();
        LongWritable value = new LongWritable();
        PairT<Long> result = new PairT<Long>(0l,0l);
        int count = 0;
        ArrayList<Long> localList = new ArrayList<Long>(65536);
        while (reader.next(key, value)) {
            localList.add(count, value.get());
            count++;
            if (count == iBuffsize - 1) {
                result = OPlus(result, mps_sum(localList));
                localList.clear();
                count =0;
            }
        }
        if (!localList.isEmpty()) {
            result = OPlus(result, mps_sum(localList));
        }

        IOUtils.closeStream(reader);
     //   readnum += count;

        return result.getFirst();
    }


    public static void main(String[] args) throws IOException {
// <input file path>
        Date startTime = new Date();
        String inputuri = args[0];
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(inputuri), conf);
        Path path = new Path(inputuri);

        long mps = ComputeMPS(conf, fs, path);
        Date end_time = new Date();
        System.out.println(" The seq MPS finished, result is "+mps +" time is " );
        System.out.println("--"
                + (end_time.getTime() - startTime.getTime()) / 1000 + " seconds.");

    }
}
